package com.yc.bigdata.flink.demo;

import com.yc.bigdata.flink.demo.data.WordCountData;
import com.yc.bigdata.flink.demo.vo.WordVO;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.MultipleParameterTool;
import org.apache.flink.util.Collector;
import org.apache.flink.util.Preconditions;

/**
 * <p></p>
 *
 * @author: YuanChilde
 * @date: 2020-02-10 9:33
 * @version: 1.0
 * Modification History:
 * Date    Author      Version     Description
 * -----------------------------------------------------------------
 * 2020-02-10 9:33    YuanChilde     1.0        新增
 */
public class DataSetWordCount {

    public static void main(String[] args) throws Exception {
        final MultipleParameterTool params = MultipleParameterTool.fromArgs(args);
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().setGlobalJobParameters(params);

        DataSet<String> text = null;
        if (params.has("input")) {
            for (String input : params.getMultiParameterRequired("input")) {
                if (text == null) {
                    text = env.readTextFile(input);
                } else {
                    text = text.union(env.readTextFile(input));
                }
            }
            Preconditions.checkNotNull(text, "Input DataSet should not be null.");
        } else {
            System.out.println("Executing WordCount example with default input data set.");
            System.out.println("Use --input to specify file input.");
            System.out.println("Executing WordCount example with default input data set.");
            System.out.println("Use --input to specify file input.");
            text = WordCountData.getDefaultTextLineDataSet(env);
        }

/*        DataSet<Tuple2<String, Integer>> counts = text.flatMap(new Tokenizer2())
                .groupBy(0).sum(1);*/
        DataSet<WordVO> counts = text.flatMap(new Tokenizer())
                .groupBy("word").reduce((wordVO, t1) -> new WordVO(wordVO.getWord(),wordVO.getFrequency()+t1.getFrequency()));

        if (params.has("output")) {
            counts.writeAsCsv(params.get("output"), "\n", " ");
            env.execute("WordCount Example");
        } else {
            System.out.println("Printing result to stdout. Use --output to specify output path.");
            counts.print();
        }
    }

    public static final class Tokenizer implements FlatMapFunction<String, WordVO>{
        @Override
        public void flatMap(String s, Collector<WordVO> collector) throws Exception {
            String[] tokens = s.toLowerCase().split("\\W+");

            // emit the pairs
            for (String token : tokens) {
                if (token.length() > 0) {
                    collector.collect(new WordVO(token, 1));
                }
            }
        }
    }

    public static final class Tokenizer2 implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            String[] tokens = value.toLowerCase().split("\\W+");
            for (String token : tokens) {
                if (token.length() > 0) {
                    out.collect(new Tuple2<>(token, 1));
                }
            }
        }
    }


}
